package com.wyw.learning.thread.collections.queue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * 各种队列的区别（举例）
 *  ArrayBlockQueue, LinkedBlockingQueue的区别就是ArrayDeque和LinkedList的区别，内部一个是可变数组，一个双链表
 *  SynchronousQueue这个队列只能一个个区，不具备存储功能
 *
 * 什么是阻塞队列
 *  简介、地位：
 *      阻塞队列是具有阻塞功能的队列，所以它首先是一个队列，其次是具有阻塞功能
 *      通常，阻塞队列的一段是生产者放数据用，另一端给消费者拿数据用。阻塞队列是线程安全的，所以生产者和消费者都可以是多线程的
 *
 * 阻塞功能：最有特色的两个带有阻塞功能的方法
 *  take()方法：获取并移除队列的头结点，一旦如果执行take的时候，队列里无数据，则阻塞，直到队列里有数据
 *  put()方法：插入元素。但是如果队列已满，那么就无法继续插入，则阻塞，直到队列里有了空闲空间
 *
 * 是否有界(容量大小)：这是一个非常重要的属性，无界队列意味着里面可以容纳非常多(Integer.MAX_VALUE，约2的31次，是一个非常大的数，可以近似认为是无限容量)
 *
 * 阻塞队列和线程池的关系：阻塞队列是线程池的重要组成部分
 *
 * ArrayBlockingQueue
 *  特点：
 *      有界
 *      指定容量
 *      公平：还可以指定是否需要保证公平，如果想保证公平，那么等待了最长时间的线程会被优先处理（队列中放的是任务），不过这会同时带来一定性能损耗
 *
 * LinkedBlockingQueue
 *  特点：
 *      无界
 *      容量Integer.MAX_VALUE
 *      内部结构：Node、两把锁
 *
 * PriorityBlockingQueue
 *  特点：
 *      支持优先级
 *      自然顺序（而不是先进先出，可以自己实现的排序规则）
 *      无界队列（容量不够会扩容）
 *      可以理解成PriorityQueue的线程安全版本
 *
 * SynchronousQueue
 *  特点：
 *      容量0
 *      因为SynchronousQueue不需要持有元素，他所做的就是直接传递，容量不是1
 *      所以效率很高
 *  注意点：
 *      没有peek等函数，因为没有头结点，容量0，没有头结点；同理没有iterate相关方法
 *      是一个极好的用来直接传递的并发数据结构
 *      SynchronousQueue是线程池Executors.newCachedThreadPool()使用阻塞队列
 *
 * DelayQueue
 *  特点：
 *      延迟队列，根据延迟时间排序
 *      元素需要实现Delayed接口，规定排序规则
 *      无界队列
 *
 * 非阻塞并发队列
 *  并发包中的非阻塞队列只有ConcurrentLinkedQueue这种
 *  顾名思义ConcurrentLinkedQueue使用链表作为数据结构
 *  使用CAS非阻塞算法实现线程安全（不具备阻塞功能）体现在p.casNext UNSAFE.compareAndSwapObject这部分
 *  适合用在对性能要求比较高的并发场景（不常用此队列）
 *          ConcurrentLinkedQueue和Atomic在高并发的时候，都会有很多自旋的，所以如果是并发特别高，性能还不如加锁
 *
 * 如何选择队列
 *  边界
 *  空间
 *  吞吐量
 *  功能
 *
 * @author Mr Wu    yewen.wu.china@gmail.com
 * <p>
 * Update History:
 * Author        Time            Content
 */
public class ArrayBlockingQueueDemo {
    public static void main(String[] args) {

        //打印顺序和实际执行顺序也许是不一样的，因为打印是要先抢到锁才行的
        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

        Interviewer r1 = new Interviewer(queue);
        Consumer r2 = new Consumer(queue);
        new Thread(r1).start();
        new Thread(r2).start();
    }
}

class Interviewer implements Runnable {

    BlockingQueue<String> queue;

    public Interviewer(BlockingQueue queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        System.out.println("10个候选人都来啦");
        for (int i = 0; i < 10; i++) {
            String candidate = "Candidate" + i;
            try {
                queue.put(candidate);
                System.out.println("安排好了" + candidate);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        try {
            queue.put("stop");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class Consumer implements Runnable {

    BlockingQueue<String> queue;

    public Consumer(BlockingQueue queue) {

        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String msg;
        try {
            while(!(msg = queue.take()).equals("stop")){
                System.out.println(msg + "到了");
            }
            System.out.println("所有候选人都结束了");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
